use crate::{Worker, WorkerSender};
use std::task::{Context, Poll};
use std::pin::Pin;
use futures_util::{
    future::Future,
};
use bitflags::bitflags;
use crate::worker::worker_core::Running;
use crate::channel::{channel, Receiver};
use crate::service::insert_service;
use std::collections::HashMap;
use std::any::Any;


bitflags! {
    struct ContextFlags: u8 {
        const STARTED =  0b0000_0001;
        const RUNNING =  0b0000_0010;
        const STOPPING = 0b0000_0100;
        const STOPPED =  0b0001_0000;
        const MB_CAP_CHANGED = 0b0010_0000;
    }
}

pub trait WorkerContext {
    fn context_part(&mut self) -> &mut ContextPart;
}

pub struct ContextPart {
    // 状态
    state: ContextFlags,
}

impl ContextPart {
    fn new() -> Self {
        ContextPart {
            state: ContextFlags::RUNNING
        }
    }

    fn stop(&mut self) {
        if self.state.contains(ContextFlags::RUNNING) {
            self.state.remove(ContextFlags::RUNNING);
            self.state.insert(ContextFlags::STOPPING);
        }
    }
}

impl<A: Worker<Context=Self>> WorkerContext for WorkerContextImpl<A> {
    fn context_part(&mut self) -> &mut ContextPart {
        &mut self.context_part
    }
}

pub struct WorkerContextImpl<A>
    where A: Worker<Context=WorkerContextImpl<A>> {
    context_part: ContextPart,
    receiver: Option<Receiver<A>>,
    // 启动参数
    pub args: Option<HashMap<String, Box<dyn Any>>>,
}

impl<A> Default for WorkerContextImpl<A>
    where A: Worker<Context=Self>
{
    fn default() -> Self {
        WorkerContextImpl::new(None)
    }
}

impl<A> WorkerContextImpl<A>
    where A: Worker<Context=Self>
{
    pub fn new(args: Option<HashMap<String, Box<dyn Any>>>) -> Self {
        WorkerContextImpl {
            context_part: ContextPart::new(),
            receiver: None,
            args,
        }
    }

    /// 协程运行通道的接收端，用来接受信息
    pub fn run(mut self, act: A) -> WorkerSender<A> {
        let (sender, receiver) = channel();
        self.receiver = Some(receiver);

        // 提交到全局发送
        // crate::worker::global_sender::insert_sender(TypeId::of::<A>(), sender.clone());
        // 接收端提交到一个协程进行接受信息并处理
        let fut = self.into_future(act);
        // 服务注册
        insert_service::<A>(sender.clone());

        tokio::spawn(fut);
        // 返回发送端
        WorkerSender {
            sender
        }
    }

    fn into_future(mut self, act: A) -> ContextFuture<A, Self>
    {
        ContextFuture {
            receiver: self.receiver.take().unwrap(),
            ctx: self,
            act,
        }
    }

    //
}


struct ContextFuture<A, C>
    where A: Worker<Context=C>,
          C: WorkerContext + Unpin
{
    receiver: Receiver<A>,
    act: A,
    ctx: C,
}

unsafe impl<A, C> Send for ContextFuture<A, C>
    where C: WorkerContext + Unpin,
          A: Worker<Context=C> {}

impl<A, C> ContextFuture<A, C>
    where C: WorkerContext + Unpin,
          A: Worker<Context=C>
{
    ///　判断是否还存活
    fn alive(&mut self) -> bool {
        if self.ctx.context_part().state.contains(ContextFlags::STOPPED) {
            false
        } else {
            !self.receiver.is_terminated()
        }
    }
}


impl<W, C> Future for ContextFuture<W, C>
    where C: WorkerContext + Unpin,
          W: Worker<Context=C>
{
    type Output = ();

    fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
        let this = self.get_mut();
        // 事件循环
        loop {
            if !this.ctx.context_part().state.contains(ContextFlags::STARTED) {
                this.ctx.context_part().state.insert(ContextFlags::STARTED);
                Worker::started(&mut this.act, &mut this.ctx);
            }

            // 处理消息
            this.receiver.poll(cx, &mut this.act);

            // 检查状态
            if this.ctx.context_part().state.contains(ContextFlags::RUNNING) {
                if !this.alive()
                    && Worker::stopping(&mut this.act, &mut this.ctx) == Running::Stop {
                    this.ctx.context_part().state = ContextFlags::STOPPED | ContextFlags::STARTED;
                    Worker::stopped(&mut this.act, &mut this.ctx);
                    return Poll::Ready(());
                }
            } else if this.ctx.context_part().state.contains(ContextFlags::STOPPING) {
                if Worker::stopping(&mut this.act, &mut this.ctx) == Running::Stop {
                    this.ctx.context_part().state = ContextFlags::STOPPED | ContextFlags::STARTED;
                    Worker::stopped(&mut this.act, &mut this.ctx);
                    return Poll::Ready(());
                } else {
                    this.ctx.context_part().state.remove(ContextFlags::STOPPING);
                    this.ctx.context_part().state.insert(ContextFlags::RUNNING);
                    continue;
                }
            } else if this.ctx.context_part().state.contains(ContextFlags::STOPPED) {
                Worker::stopped(&mut this.act, &mut this.ctx);
                return Poll::Ready(());
            }

            return Poll::Pending;
        }
    }
}


impl<A, C> Drop for ContextFuture<A, C>
    where C: WorkerContext + Unpin,
          A: Worker<Context=C> {
    fn drop(&mut self) {
        if self.alive() {
            // 设置关闭
            self.ctx.context_part().stop();
            // 再次运行
            let waker = futures_util::task::noop_waker();
            let mut ctx = futures_util::task::Context::from_waker(&waker);
            let _ = Pin::new(self).poll(&mut ctx);
        }
    }
}


